package com.wuwangfu.state;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: jcshen
 * @Date: 2023-03-03
 * @PackageName: com.wuwangfu.state
 * @ClassName: ListStated
 * @Description: ListState（Value 是一个list集合）
 * @Version: 1.0.0
 *
 * keyBy之后，用来存储K-V类型的状态，叫做KeyedState
 *
 */
public class ListStated {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        //
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000));
        /**
         * user1,A
         * user1,C
         * user2,B
         */
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple2<String, String>> flated = line.flatMap(new FlatMapFunction<String, Tuple2<String, String>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, String>> out) throws Exception {
                String[] field = value.split(",");
                out.collect(Tuple2.of(field[0], field[1]));
            }
        });
        //分组
        KeyedStream<Tuple2<String, String>, String> keyed = flated.keyBy(t -> t.f0);
        //
        keyed.process(new KeyedProcessFunction<String, Tuple2<String, String>, Tuple2<String, List<String>>>() {

            private transient ListState<String> listState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //定义一个状态描述器
                ListStateDescriptor<String> listDesc = new ListStateDescriptor<>("list-state", String.class);
                //初始化状态或恢复zhuangt
                listState = getRuntimeContext().getListState(listDesc);
            }

            @Override
            public void processElement(Tuple2<String, String> input, Context context, Collector<Tuple2<String, List<String>>> out) throws Exception {
                //输入值
                String action = input.f1;
                //添加状态值
                listState.add(action);
                Iterable<String> iterator = listState.get();
                List<String> events = new ArrayList<>();
                for (String name:iterator){
                    events.add(name);
                }
                out.collect(Tuple2.of(input.f0,events));
            }
        }).print();

        env.execute();
    }
}
